package rdd01.part

import org.apache.spark.rdd.RDD
import org.apache.spark.{Partitioner, SparkConf, SparkContext}

object RDD_Part {
  //自己创建分区器
  def main(args: Array[String]): Unit = {
    val sparkconf=new SparkConf().setMaster("local[*]").setAppName("persis")
    val sc=new SparkContext(sparkconf)
    val rdd=sc.makeRDD(List(
      ("nba","xsssssssss"),
      ("cba","xsssssssss"),
      ("wnba","xsssssssss"),
      ("nba","xsssssssss"),
      ("nba","xsssssssss")
    ),3)
    val partRDD:RDD[(String,String)]=rdd.partitionBy(new MyPatitioner)
    partRDD.saveAsTextFile("partRDD")
    sc.stop()
  }
  /**自定义分区器,需要重写其中的两个方法
   * */
  class MyPatitioner extends Partitioner{
    //分区的数量
    override def numPartitions: Int = 3
   //返回数据的分区索引，根据数据的key值返回数据所在的分区索引（从0开始）
    override def getPartition(key: Any): Int ={
//      if(key=="nba"){
//        0
//      }else if(key=="cba"){
//        1
//      }else if(key=="wnba"){
//        3
//      }else{
//        2
//      }
      //采用模式匹配规则
      key match{
        case "nba"=>0
        case "wnba"=>1
        case "cba"=>2
        case _=>3
      }
    }
  }

}
